// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_PARALLEL_SCANNER_HPP_
#define _XBASE_PARALLEL_SCANNER_HPP_

#include <map>
#include <vector>
#include <string>

#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>

#include "tablet_scanner.hpp"
#include "location_cache.hpp"
#include "io_service_pool.hpp"
#include "session_manager.hpp"

namespace xbase
{
	class parallel_scanner
		: public async_module_base<io_service_pool::user_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::user_level, io_service_pool::xbase_level> base_type;
		typedef parallel_scanner this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef boost::asio::ip::tcp::endpoint endpoint_type;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(row_result::ptr_type, const error_code&)> handler_type;

	public:
		enum scan_state
		{
			not_assigned = 0,
			scanning,
			finished
		};

		struct scan_state_map_value
		{
			std::string m_end_row;
			scan_state m_state;
			bool m_start_row_got;
		};

		typedef std::map<std::string, scan_state_map_value> scan_state_map;

	public:
		parallel_scanner(io_service_pool &pool, session_manager &sm, location_cache &lc,
			const scan_info &si, handler_type handler);

		void start();
	private:
		void on_get_meta(location_cache_value meta, const error_code& ec);

		void on_get_next_row(tablet_scanner_ptr tablet_scanner_p, bool start_row_got,
			uint64_t got_meta_time, row_result::ptr_type row_ptr, const error_code &ec);

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		location_cache &m_location_cache;
		session_manager &m_session_manager;

		bool m_started;
		tablet_scan_info m_tablet_scan_info;

		handler_type m_handler;

		scan_state_map m_scan_state_map;

		io_service_pool &m_io_service_pool;

		std::string m_table_name;
		std::string m_start_row;
		std::string m_end_row;
	};

	typedef parallel_scanner::ptr_type parallel_scanner_ptr;
}

#endif

